import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * @ Auther: o2o-rd-0008
  * @ Date:   2019/10/8 16:47
  * @ Param:  ${PARAM}
  * @ Description: 
  */
object modify_price_9 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.set("spark.debug.maxToStringFields", "500")
    conf.setMaster("local[*]")
    conf.set("es.nodes", "192.168.1.157")
    conf.set("es.port", "9200")
    conf.set("cluster.name", "O2OElastic")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.set("spark.sql.caseSensitive", "true")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    sc.setLogLevel("WARN")
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")

    import org.elasticsearch.spark._

    val values: RDD[String] = sc.esJsonRDD(s"2020_tmall/tmall_2020_3",
      """
        |{
        |  "query": {
        |    "term": {
        |      "shopName.keyword": {
        |        "value": "拼腾旗舰店"
        |      }
        |    }
        |  }
        |}
      """.stripMargin).values

    val frame: DataFrame = sqlContext.read.json(values).where("cast(priceText as double)>300")
    frame.selectExpr("priceText","salesAmount").show(false)


    frame.createOrReplaceTempView("frame")
    //frame.write.json("s3a://private/private_han/TM/2019/9/modify_taobao")
    val frame1 =  sqlContext.sql(
      """
        |select
        |*,
        |(sellCount*price1) as sales
        |from
        |(select
        |*,
        |cast(priceText as double)-300 as price1
        |from
        |frame)
      """.stripMargin).drop("priceText","salesAmount")
      .withColumnRenamed("price1","priceText")
      .withColumnRenamed("sales","salesAmount")

    frame1.selectExpr("priceText","salesAmount").show(false)

    val result = frame1.toJSON.rdd.map(line => {
      val lines = JSON.parseObject(line)
      val evaluates = lines.getOrDefault("evaluates", "-1").toString
      val Base_Info = lines.getOrDefault("Base_Info", "-1").toString
      var ev = "-1"
      var str = new JSONObject
      if (evaluates.equals("-1")) {
        ev = "{\"fuyi\":\"-1\"}"
      } else {
        str = JSON.parseObject(evaluates)
      }
      if (!ev.contains("fuyi")) {
        lines.put("evaluates", str)
      } else {
        val evs = JSON.parseObject(ev)
        lines.put("evaluates", evs)
      }
      lines.put("Base_Info",Base_Info)
      lines
    })
    result.saveToEs(s"2020_tmall/tmall_2020_3", Map("es.mapping.id" -> "good_id"))

  }

  def min_price(price1:String,price2:String):String={
    var p = 0.0
    val p1 = price1.toDouble
    val p2 = price2.toDouble
    if (p1<=p2){
      p = p1
    }else{
      p = p2
    }
    p.toString
  }
}
